package doris;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;

import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
 *  读取socket数据，写入doris
 */

public class Demo {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<String> stream = env.socketTextStream("sjz01", 44444);

        DataStream<UserLog> userLogs = stream.map(new MapFunction<String, UserLog>() {
            @Override
            public UserLog map(String value) throws Exception {
                String[] parts = value.split(",");
                return new UserLog(Integer.parseInt(parts[0]), parts[1], Long.parseLong(parts[2]));
            }
        });

        userLogs.addSink(JdbcSink.sink(
                "INSERT INTO demo.user_logs (user_id, activity, timestamp) VALUES (?, ?, ?)",
                new JdbcStatementBuilder<UserLog>() {
                    @Override
                    public void accept(PreparedStatement ps, UserLog log) throws SQLException {
                        ps.setInt(1, log.userId);
                        ps.setString(2, log.activity);
                        ps.setLong(3, log.timestamp);
                    }
                },
                JdbcExecutionOptions.builder()
                        .withBatchSize(1000)
                        .build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withUrl("jdbc:mysql://sjz02:9030/demo")
                        .withDriverName("com.mysql.cj.jdbc.Driver")
                        .withUsername("root")
                        .withPassword("Sjz@0626")
                        .build()
        ));

        env.execute("Flink Doris Demo");
    }


    public static class UserLog {
        public int userId;
        public String activity;
        public long timestamp;

        public UserLog() {}

        public UserLog(int userId, String activity, long timestamp) {
            this.userId = userId;
            this.activity = activity;
            this.timestamp = timestamp;
        }

        @Override
        public String toString() {
            return "UserLog{" +
                    "userId=" + userId +
                    ", activity='" + activity + '\'' +
                    ", timestamp=" + timestamp +
                    '}';
        }
    }
}
